package com.ow.dva.module.base.service.impl;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.ow.dva.module.base.entity.KafkaMessage.KafkaMessage;
import com.ow.dva.module.base.service.KafkaService;
import com.ow.dva.module.order.service.OrderService;
import com.ow.dva.module.passingRecord.entity.PassingRecord;
import com.ow.dva.module.passingRecord.service.PassingRecordService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

@Service
public class KafkaServiceImpl implements KafkaService {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;
    @Resource
    private OrderService orderService;
    @Resource
    private PassingRecordService passingRecordService;

    public Logger logger = LoggerFactory.getLogger(this.getClass());

    @Override
    public void sendMessage(Object message) {
         String msg = JSONUtil.toJsonStr(message);
        logger.info("Send to topic : [dva-client] ==> " + msg);
        kafkaTemplate.send("dva-client",msg);
    }

    @Override
    public void sendMessage(String topic,Object message) {
        String msg = JSONUtil.toJsonStr(message);
        logger.info("Send to kafka : [" + topic + "] ==> " + msg);
        kafkaTemplate.send(topic,msg);
    }

    @Override
    public void sendMessage(Integer type, String topic, Object bean) {
        Map map = new HashMap();
        map.put("type",type);
        String beanStr = JSONUtil.toJsonStr(bean);
        map.put("beanStr",beanStr);
        String msg = JSONUtil.toJsonStr(map);
        msg = msg.replace("\\\\","");
        logger.info("Send to kafka : [" + topic + "] ==> " + msg);
        ListenableFuture<SendResult<String,String>> listenableFuture = kafkaTemplate.send(topic,msg);
        listenableFuture.addCallback(//添加成功发送消息的回调和失败的回调
                result -> logger.info("send message to {} success",topic),
                ex -> logger.info("send message to {} failure,error message:{}",topic,ex.getMessage()));
//        System.out.println(listenableFuture);
    }

    //    @KafkaListener(topics = "#{'${dva.kafka-topic}'.split(',')}")
    @KafkaListener(topics = "dva-client-server")
    public void kafkaListener(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            readMessage(record);
            logger.info("Receive to kafka [topic="+record.topic()+",partition="+record.partition()+",offset="+record.offset()+"] ==> " + message);

        }
    }

    /**
     *
     * @param record
     */
    public void readMessage(ConsumerRecord<?, ?> record){
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            String msg = JSONUtil.toJsonStr(message);
            KafkaMessage kafkaMessageBean = JSON.parseObject(msg,KafkaMessage.class);
            Integer type = kafkaMessageBean.getType();
            String tTopic = kafkaMessageBean.gettId();
            if (type == 1){
                int result = passingRecordService.getJsonRecord(kafkaMessageBean.getBeanStr());
                if (result>0){
//                    sendMessage(tTopic,message);
                    System.out.println("收到过车记录保存成功");
                }
            }else if (type == 2){
                int result = orderService.getJsonOrder(kafkaMessageBean.getBeanStr());
                System.out.println("收到订单保存成功");
                if (result>0){
//                    sendMessage(tTopic,message);
                }
            }else if (type == 3){//服务器接收到了终端发来的场内支付订单
                int result = orderService.getJsonOrder(kafkaMessageBean.getBeanStr());
                if (result>0){
//                    sendMessage(tTopic,message);
                }
            }else {

            }
        }

    }
}
